package com.txl.cn.spark03

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

/**
  * Created by txl on 2017/12/30.
  *需求：根据访问日志中的ip地址，获取到对应的归属地，并且按照归属地进行计数统计，然后把结果写入到mysql中
  * 有一个中间的应用库
  * 获取访问日志中的每一个ip地址，
  * 然后把ip地址转换成十进制   数字    方法1
  * 拿这一个数字去ip规则库中进行匹配   匹配搜索方法
  * 匹配完成之后，可以获取到省份   把省份和1组装起来（省份，1）
  * 剩下的统计聚合，就是一个wordcount
  * 得到的结果，现在是是把结果数据写入到mysql中   方法3
  */
object LocalIP {
  def ip2Long(ip: String): Long = {
    val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length) {
      ipNum = fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }
  def search(ipNum:Long,ipRules:Array[(Long,Long,String)]):Int={
    var start=0
    var end=ipRules.length-1
    while(start<=end){
      var middle=(start+end)/2
      if(ipNum >=ipRules(middle)._1 && ipNum<=ipRules(middle)._2){
            return middle
      }else if(ipNum<ipRules(middle)._1){
        end=middle-1
      }else{
        start=middle+1
      }
    }
    -1
  }


  def main(args: Array[String]): Unit = {
   val ipNum=ip2Long("115.120.4.15")
    val lines = Source.fromFile("data/ip.txt").getLines()
    val ipRules: Array[(Long, Long, String)] = lines.map({
      line =>
        val rules = line.split("\\|")
        val start = rules(2).toLong
        val end = rules(3).toLong
        val province = rules(6)
        (start, end, province)
    }).toArray
    val index = search(ipNum, ipRules)
    if(index != -1)  println(ipRules(index))  else  println("unknown")



    val conf = new SparkConf().setAppName("ip").setMaster("local")
    val sc=new SparkContext(conf)
    val logs = sc.textFile("data/access.log")
    val res: RDD[((Long, Long, String), Int)] = logs.map({
      t =>
        val ip = t.split("[|]")(0)
        val i = ip2Long(ip)
        val index1 = search(i, ipRules)
        val province = ipRules(index1)
        (province, 1)
    })
    val data: RDD[((Long, Long, String), Int)] = res.reduceByKey(_+_)
    var conn: Connection = null
    var pstm: PreparedStatement = null
    try {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/scott", "root", "123456")
      pstm = conn.prepareStatement("insert into access_log values (?,?)")
      // 因为拿到的每一个元素是一个迭代器，所以这里还需要循环迭代
      data.foreachPartition({
        t =>
           t.foreach({
             x=>
               pstm.setString(1,x._1._3)
               pstm.setInt(2,x._2)
           })
          pstm.executeUpdate()
      })
    } catch {
      case e: Exception => println(e.printStackTrace())
    }
    finally {
      // 关闭连接
      if (pstm != null)
        pstm.close()
      if (conn != null) conn.close()
    }





    }

}
